package com.yjd.bi.report.service.impl;

import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.annotation.Resource;

import org.apache.log4j.Logger;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;

import com.yjd.bi.report.enums.BiDataChangeRecordStatusEunm;
import com.yjd.bi.report.enums.BiDataChangeRecordTypeEnum;
import com.yjd.bi.report.model.BiDataChangeRecordModel;
import com.yjd.bi.report.model.BiDatabaseInfoModel;
import com.yjd.bi.report.model.BiEtlFromtableTotableModel;
import com.yjd.bi.report.model.BiTableInfoModel;
import com.yjd.bi.report.service.IBiDataChangeRecordService;
import com.yjd.bi.report.service.IBiDatabaseInfoService;
import com.yjd.bi.report.service.IBiEtlFromtableTotableService;
import com.yjd.bi.report.service.IBiTableInfoService;
import com.yjd.bi.report.service.IDataChangeRecordDisposeService;
import com.yjd.bi.report.util.BiUtil;
import com.yjd.comm.base.concurrent.task.ITaskHandler;
import com.yjd.comm.base.concurrent.task.ITaskItem;
import com.yjd.comm.base.concurrent.task.TaskDisposeUtil;
import com.yjd.comm.base.enums.DbWREnums;
import com.yjd.comm.util.FrameUtil;

/**
 * 
 * <b>description</b>：本类中不通过spring来控制事务 <br>
 * <b>time</b>：2016-4-14下午4:04:50 <br>
 * <b>author</b>： ready likun_557@163.com
 */
@Service("dataChangeRecordDisposeService")
public class DataChangeRecordDisposeService implements
		IDataChangeRecordDisposeService {
	private Logger logger = Logger
			.getLogger(BiDataChangeRecordServiceImpl.class);

	/**
	 * 根据BiDataChangeRecordModel将数据迁移至bi库，内部发生异常表示失败
	 * 
	 * @param dcr
	 * @return
	 * @throws Exception
	 */
	public boolean dispose(BiDataChangeRecordModel dcr) throws Exception {
		boolean result = false;
		if (dcr != null
				&& !BiDataChangeRecordStatusEunm.SUCCESS.getValue().equals(
						dcr.getStatus()) && dcr.getDatabase_info_id() != null
				&& dcr.getTable_name() != null && dcr.getRef_id() != null) {
			long curtime = FrameUtil.getTime();
			String message = "处理成功!";
			try {
				// 处理数据
				result = this.disposeIn(dcr);
			} catch (Exception e) {
				this.logger.error("业务数据迁移至bi库出错:" + e.getMessage(), e);
				message = e.getMessage();
				result = false;
			}
			// 处理之后，修改dcr状态
			this.disposeAfter(dcr, result, curtime, message);
		}
		return result;
	}

	private void disposeAfter(BiDataChangeRecordModel dcr,
			boolean disposeSuccess, long curtime, String message)
			throws Exception {
		dcr.setMessage(message);
		dcr.setActual_dispose_time(curtime);
		dcr.setStatus(disposeSuccess ? BiDataChangeRecordStatusEunm.SUCCESS
				.getValue() : BiDataChangeRecordStatusEunm.FAIL.getValue());
		if (!disposeSuccess) {
			dcr.setFailure(dcr.getFailure() + 1);
			// 如果处理失败，延迟10秒处理
			dcr.setDispose_time(curtime + 10);
		}
		this.biDataChangeRecordService.update(dcr);
	}

	/**
	 * 根据变更日志，将业务库数据迁移到bi库中
	 * 
	 * @param dcr
	 * @return
	 * @throws Exception
	 */
	@SuppressWarnings("unchecked")
	private boolean disposeIn(BiDataChangeRecordModel dcr) throws Exception {
		// 业务库
		BiDatabaseInfoModel fromDb = this.biDatabaseInfoService.getModelById(
				dcr.getDatabase_info_id(), DbWREnums.READ);
		if (fromDb == null && logger.isInfoEnabled()) {
			logger.info("数据迁移,BiDataChangeRecordModel:" + FrameUtil.json(dcr)
					+ ",关联的数据库不存在!");
			return false;
		} else {
			// 业务数据
			BiTableInfoModel fromTable = this.biTableInfoService.getModelOne(
					FrameUtil.newHashMap("database_info_id", fromDb.getId(),
							"table_name_ignore_case", dcr.getTable_name()),
					DbWREnums.READ);
			if (fromTable == null && logger.isInfoEnabled()) {
				logger.info("数据迁移,BiDataChangeRecordModel:"
						+ FrameUtil.json(dcr) + ",关联的表不存在!");
				return false;
			}
			// 数据抽取中原表和目标表关联信息
			List<BiEtlFromtableTotableModel> biEtlFromtableTotableModels = this.biEtlFromtableTotableService
					.getModelList(
							FrameUtil.newHashMap("from_table_id",
									fromTable.getId()), DbWREnums.READ);
			if (biEtlFromtableTotableModels != null
					&& !biEtlFromtableTotableModels.isEmpty()) {
				Map<Long, BiTableInfoModel> toTablesMap = this.biTableInfoService
						.getModelMapByIds(FrameUtil.getBeanFieldValues(
								biEtlFromtableTotableModels, "to_table_id"),
								DbWREnums.READ);
				Collection<BiTableInfoModel> toTables = toTablesMap.values();
				for (BiTableInfoModel toTable : toTables) {
					BiDatabaseInfoModel toDb = this.biDatabaseInfoService
							.getModelById(toTable.getDatabase_info_id(),
									DbWREnums.READ);
					if (toDb != null) {
						final String deleteSql = String.format(
								"delete from %s where %s = ?",
								toTable.getTable_name(),
								toTable.getPk_field_name());
						final Object[] deleteArgs = new Object[] { dcr
								.getRef_id() };
						if (BiDataChangeRecordTypeEnum.DELETE.getValue()
								.equals(dcr.getType())) {
							BiUtil.getJdbcTemplate(toDb).update(deleteSql,
									deleteArgs);
						} else {
							// 在业务库查询记录
							List<Map<String, Object>> rows = BiUtil
									.getJdbcTemplate(fromDb)
									.queryForList(
											String.format(
													"select * from %s where %s = ?",
													fromTable.getTable_name(),
													fromTable
															.getPk_field_name()),
											dcr.getRef_id());
							// 获取表的列明
							List<String> columnNames = BiUtil.getColumnNames(
									toTable.getTable_name(),
									BiUtil.getJdbcTemplate(toDb));
							if (rows != null && !rows.isEmpty()) {
								Map<String, Object> row = rows.get(0);
								StringBuilder colSql = new StringBuilder();
								StringBuilder colParam = new StringBuilder();
								List<Object> colParamValues = FrameUtil
										.newArrayList();
								for (String column : columnNames) {
									colSql.append("`").append(column)
											.append("`,");
									colParam.append("?").append(",");
									colParamValues.add(row.get(column));
								}
								final String insertSql = String.format(
										"insert into %s (%s) values (%s)",
										toTable.getTable_name(), colSql
												.substring(0,
														colSql.length() - 1),
										colParam.substring(0,
												colParam.length() - 1));
								final Object[] insertArgs = colParamValues
										.toArray();
								// 新增和修改：先删除输入，然后执行插入
								BiUtil.execute(toDb,
										new BiUtil.ITransExecute<Boolean>() {
											@Override
											public Boolean execute(
													JdbcTemplate jt)
													throws Exception {
												// 先执行删除
												jt.update(deleteSql, deleteArgs);
												// 执行插入
												jt.update(insertSql, insertArgs);
												return true;
											}
										});
							}
						}
					}
				}
			}
		}
		return true;
	}

	/**
	 * 将bi库处理完毕以及处理失败的数据移到日志表中
	 * 
	 * @throws Exception
	 */
	@SuppressWarnings({ "unchecked" })
	public void moveDisposedToLog() throws Exception {
		while (true) {
			Map<Object, Object> paramMap = new HashMap<Object, Object>();
			paramMap.put("whereSql", "(a.status="
					+ BiDataChangeRecordStatusEunm.SUCCESS.getValue()
					+ " or max_failure<=failure)");
			paramMap.put("rowstartindex", 0);
			paramMap.put("rowendindex", 1001);
			paramMap.put("sort", "id asc");
			List<BiDataChangeRecordModel> list = this.biDataChangeRecordService
					.getModelList(paramMap, DbWREnums.WRITE);
			// 去处最后一条，防止数据被迁移完毕之后，数据库重启之后，主键重复
			if (list != null && list.size() >= 2) {
				list.remove(list.size() - 1);
				this.biDataChangeRecordService.moveDataToDataLog(FrameUtil
						.getBeanFieldValues(list, "id"));
			} else {
				break;
			}
		}
	}

	@Resource
	private IBiEtlFromtableTotableService biEtlFromtableTotableService;

	@Resource
	private IBiDatabaseInfoService biDatabaseInfoService;
	@Resource
	private IBiDataChangeRecordService biDataChangeRecordService;
	@Resource
	private IBiTableInfoService biTableInfoService;
}
